package cc.ab01;

import java.io.IOException;
import java.io.Reader;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

import org.apache.log4j.Logger;

import oracle.to.mysql.utils.SqlUtils;
/**
 *  ab01转入后，将重复数据放入ab01_temp,将其aab001设置为正式库的aab001
 *  之后只需将数据更新即可
 */
public class A1_AB01Insert_temp {
	
	private static String mysqlInsertSql;
	
	private static int mysqlcolsSql;
	private static Logger log = Logger.getLogger(" ");
	private static int poolsize = 20;//线程池数量
	private static int pgsize = 500;//单个线程执行的sql数量,oracle分页条数
	private static String ORACLE_TABLE = "AB01_CC";
	private static String ORACLE_TABLE_ORDERBY = "aab001";
	private static String MYSQL_IDENTIFY = "DXP_CB20_AAB001";//mysql标识字段，知道哪些数据是导入的
	private static String MYSQL_TABLE = "AB01";
	
	private static String mysqlInsertSqlTemp;
	private static String MYSQL_TABLE_TEMP = "AB01_TEMP";
	private static final String jyt_url = "jdbc:mysql://10.20.0.32:3306/business?user=jlrcbusiness&password=jlrcbusiness&useUnicode=true&useSSL=false&characterEncoding=UTF8";

	static {
		try {
			Class.forName("oracle.jdbc.driver.OracleDriver");
			Class.forName("com.mysql.jdbc.Driver");
		} catch (ClassNotFoundException e) {
			log.error("驱动加载异常：" + e.getMessage());
		}
	}
	
	public static void main(String[] args) throws SQLException, IOException {
		long time = System.currentTimeMillis();
		Connection connOracle = DriverManager.getConnection(SqlUtils.ORACLE_URL, SqlUtils.ORACLE_USER, SqlUtils.ORACLE_PWD);
		getInsertSql(connOracle);
		int count = getCount(connOracle, ORACLE_TABLE);
		connOracle.close();
		ExecutorService pool = Executors.newFixedThreadPool(poolsize);
		int zheng = count / pgsize;
		int yu = count % pgsize;
		A1_AB01Insert_temp r = new A1_AB01Insert_temp();
		for (int i = 1; i <= zheng; i++) {
			pool.execute(r.new TransRunnable((i - 1) * pgsize + 1, i * pgsize));
		}
		pool.execute(r.new TransRunnable(count - yu, count));
		pool.shutdown();
		try {
			while (!pool.awaitTermination(1, TimeUnit.SECONDS)) {
				// awaitTermination返回false即超时会继续循环，返回true即线程池中的线程执行完成主线程跳出循环往下执行，每隔10秒循环一次
			}
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
		log.info("oracle表："+ ORACLE_TABLE+",条数："+ count);
		log.info("线程池总数："+ poolsize+","+"子线程数量："+ (zheng+1));
		log.info("子线程执行条数："+ pgsize);
		log.info("Mysql标识字段: "+MYSQL_IDENTIFY);
		log.info("全部执行完成，mysql:"+MYSQL_TABLE+"，耗时:"+(System.currentTimeMillis()-time)+"ms");
	}

	private static int getCount(Connection conn, String table) throws SQLException {
		long time = System.currentTimeMillis();
		String sql = "select count(1) from " + table.toUpperCase();
		Statement stmt = conn.createStatement(ResultSet.TYPE_SCROLL_INSENSITIVE, ResultSet.CONCUR_UPDATABLE);
		ResultSet rs = stmt.executeQuery(sql);
		int count = 0;
		if (rs.next()) {
			count = rs.getInt(1);
		}
		rs.close();
		stmt.close();
		log.info("查询总数:" + count + "条,耗时：" + (System.currentTimeMillis() - time));
		return count;
	}


	private static void getInsertSql(Connection connOracle) throws SQLException {
		Statement stmt = connOracle.createStatement();
		String querySql = SqlUtils.getSql(ORACLE_TABLE,null,ORACLE_TABLE_ORDERBY,0, 1);
		ResultSet rs = stmt.executeQuery(querySql);
		ResultSetMetaData mt = rs.getMetaData();
		int cols = mt.getColumnCount();
		StringBuffer sb = new StringBuffer(128);
		sb.append("insert into ");
		sb.append(SqlUtils.MYSQL_DB);
		sb.append(".");
		sb.append(MYSQL_TABLE);
		sb.append("(");
		for(int i=1;i<=cols;i++){
			String name = mt.getColumnName(i);
			if (i != 1) {
				sb.append(",");
			}
			if(name.equalsIgnoreCase("rn")){
				sb.append(MYSQL_IDENTIFY);
			}else{
				sb.append(name);
			}
		}
		sb.append(") values (");
		for (int c = 1; c <= cols; c++) {
			if (c != 1) {
				sb.append(",");
			}
			sb.append("?");
		}
		sb.append(")");
		rs.close();
		stmt.close();
		mysqlInsertSql = sb.toString();
		mysqlcolsSql = cols;//最后一项rn
		mysqlInsertSqlTemp = sb.toString().replace("."+MYSQL_TABLE+"(", "."+MYSQL_TABLE_TEMP+"(");
	}
	
	class TransRunnable implements Runnable {
		int start;
		int end;

		public TransRunnable(int start, int end) {
			this.start = start;
			this.end = end;
		}
		private int countJYT(Connection conn, String table,String aab004){
			try {
				String sql = "select count(1) from " + table.toUpperCase()+" where aab004='"+aab004+"'";
				Statement stmt = conn.createStatement(ResultSet.TYPE_SCROLL_INSENSITIVE, ResultSet.CONCUR_UPDATABLE);
				ResultSet rs = stmt.executeQuery(sql);
				int count = 0;
				if (rs.next()) {
					count = rs.getInt(1);
				}
				rs.close();
				stmt.close();
				return count;
			} catch (SQLException e) {
				e.printStackTrace();
			}
			return 0;
		}
		private String aab001JYT(Connection conn, String table,String aab004){
			try {
				String sql = "select aab001 from " + table.toUpperCase()+" where aab004='"+aab004+"'";
				Statement stmt = conn.createStatement(ResultSet.TYPE_SCROLL_INSENSITIVE, ResultSet.CONCUR_UPDATABLE);
				ResultSet rs = stmt.executeQuery(sql);
				String aab001 = null;
				if (rs.next()) {
					aab001 = rs.getString(1);
				}
				rs.close();
				stmt.close();
				return aab001;
			} catch (SQLException e) {
				e.printStackTrace();
			}
			return null;
		}
		@Override
		public void run() {
			Connection connOracle = null;
			Connection connMysql = null;
			Connection connMysqlJYT = null;
			long time = System.currentTimeMillis();
			Statement stmt = null;
			ResultSet rs = null;
			int error = 0;
			try {
				connOracle = DriverManager.getConnection(SqlUtils.ORACLE_URL, SqlUtils.ORACLE_USER, SqlUtils.ORACLE_PWD);
				connMysql = DriverManager.getConnection(SqlUtils.getMySqlUrl());
				connMysqlJYT = DriverManager.getConnection(jyt_url);
				stmt = connOracle.createStatement();
				rs = stmt.executeQuery(SqlUtils.getSql(ORACLE_TABLE,null,ORACLE_TABLE_ORDERBY,start, end));
				while (rs.next()) {
					String aab004 = rs.getString("aab004");
					PreparedStatement stmt2 = null;
					if(countJYT(connMysqlJYT, MYSQL_TABLE, aab004) != 0){
						String aab001jyt = aab001JYT(connMysqlJYT, MYSQL_TABLE, aab004);
						stmt2 = connMysql.prepareStatement(mysqlInsertSqlTemp);
						for (int c = 1; c <= mysqlcolsSql; c++) {
							Object o = rs.getObject(c);
							if(c == 1){
								stmt2.setObject(c, aab001jyt);
							}else{
								if(mysqlcolsSql==c){
									stmt2.setObject(c, "财务数据导入");//标识字段
								}else{
									if (o != null) {
										if (o instanceof oracle.sql.TIMESTAMP) {
											oracle.sql.TIMESTAMP ts = (oracle.sql.TIMESTAMP) o;
											o = new java.sql.Timestamp(ts.timestampValue().getTime());
										} else if (o instanceof java.util.Date) {
											o = new java.sql.Timestamp(((java.util.Date) o).getTime());
										} else if (o instanceof java.sql.Clob) {
											oracle.sql.CLOB clob = (oracle.sql.CLOB) o;
											Reader red = clob.getCharacterStream();
											char[] crs = new char[10240];
											String tmp = "";
											int len;
											while ((len = red.read(crs)) != -1) {
												tmp += new String(crs, 0, len);
											}
											red.close();
											o = new String(tmp);
										}
									}
									stmt2.setObject(c, o);
								}
							}
						}
					}else{
						 stmt2 = connMysql.prepareStatement(mysqlInsertSql);
						 for (int c = 1; c <= mysqlcolsSql; c++) {
							 Object o = rs.getObject(c);
							 if(mysqlcolsSql==c){
								 stmt2.setObject(c, "财务数据导入");//标识字段
							 }else{
								 if (o != null) {
									 if (o instanceof oracle.sql.TIMESTAMP) {
										 oracle.sql.TIMESTAMP ts = (oracle.sql.TIMESTAMP) o;
										 o = new java.sql.Timestamp(ts.timestampValue().getTime());
									 } else if (o instanceof java.util.Date) {
										 o = new java.sql.Timestamp(((java.util.Date) o).getTime());
									 } else if (o instanceof java.sql.Clob) {
										 oracle.sql.CLOB clob = (oracle.sql.CLOB) o;
										 Reader red = clob.getCharacterStream();
										 char[] crs = new char[10240];
										 String tmp = "";
										 int len;
										 while ((len = red.read(crs)) != -1) {
											 tmp += new String(crs, 0, len);
										 }
										 red.close();
										 o = new String(tmp);
									 }
								 }
								 stmt2.setObject(c, o);
							 }
						 }
					}
					try {
						stmt2.execute();
					} catch (Exception e) {
						log.error("插入mysql异常,oracle主键："+e.getMessage().replace("Duplicate entry '", "").replace("' for key 'PRIMARY'", "")+",sql:"+stmt2.toString());
						error++;
					}
					stmt2.close();
				}
			} catch (Exception e) {
				log.error("子线程异常："+e.getMessage());
			} finally {
				try {
					if (stmt != null) {
						stmt.close();
					}
				} catch (SQLException e) {
				}
				try {
					if (rs != null) {
						rs.close();
					}
				} catch (SQLException e) {
				}
				try {
					if (connMysql != null) {
						connMysql.close();
					}
				} catch (SQLException e) {
				}
				try {
					if (connOracle != null) {
						connOracle.close();
					}
				} catch (SQLException e) {
				}
				try {
					if (connMysqlJYT != null) {
						connMysqlJYT.close();
					}
				} catch (SQLException e) {
				}
				log.info("子线程完成，start:" + start + ",end：" + end + ",用时：" + (System.currentTimeMillis() - time) +"ms,error:"+error);
			}
		}

	}

}